package com.gjy.kafka.sp.service;

import com.gjy.kafka.sp.entity.MessageEntity;
import com.gjy.kafka.sp.entity.MessageTrace;
import com.gjy.kafka.sp.mapper.MessageTraceMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.time.LocalDateTime;
import java.util.Objects;

/**
 * 消息轨迹服务
 * 记录消息的发送和消费轨迹
 *
 * @author gjy
 * @version 1.0
 * @since 2025-09-07 10:42:50
 */
@Service
public class MessageTraceService {
    private static final Logger log = LoggerFactory.getLogger(MessageTraceService.class);

    private final MessageTraceMapper messageTraceMapper;

    public MessageTraceService(MessageTraceMapper messageTraceMapper) {
        this.messageTraceMapper = messageTraceMapper;
    }

    /**
     * 记录消息发送前的轨迹
     *
     * @param message 消息实体
     * @param topic   主题
     * @return 消息轨迹ID
     */
    @Transactional(rollbackFor = Exception.class)
    public Long recordBeforeSend(MessageEntity message, String topic) {
        Objects.requireNonNull(message, "消息实体不能为空");

        MessageTrace trace = new MessageTrace();
        trace.setMessageId(message.getMessageId());
        trace.setTopic(topic);
        trace.setBusinessType(message.getBusinessType());
        trace.setBusinessId(message.getBusinessId());
        trace.setContent(message.getContent());
        trace.setSendStatus(0); // 待发送
        trace.setCreateTime(LocalDateTime.now());
        trace.setUpdateTime(LocalDateTime.now());

        messageTraceMapper.insert(trace);
        log.info("记录消息发送前轨迹，消息ID：{}，轨迹ID：{}", message.getMessageId(), trace.getId());
        return trace.getId();
    }

    /**
     * 记录消息发送成功的轨迹
     *
     * @param messageId 消息ID
     * @param partition 分区
     * @param offset    偏移量
     */
    @Transactional(rollbackFor = Exception.class)
    public void recordSendSuccess(String messageId, int partition, long offset) {

        MessageTrace trace = messageTraceMapper.selectByMessageId(messageId);
        if (trace == null) {
            log.warn("未找到消息轨迹，消息ID：{}", messageId);
            return;
        }

        trace.setSendTime(LocalDateTime.now());
        trace.setSendStatus(1); // 发送成功
        trace.setMqPartition(partition);
        trace.setOffset(offset);
        trace.setUpdateTime(LocalDateTime.now());

        messageTraceMapper.updateById(trace);
        log.info("记录消息发送成功轨迹，消息ID：{}", messageId);
    }

    /**
     * 记录消息发送失败的轨迹
     *
     * @param messageId 消息ID
     * @param errorMsg  错误信息
     */
    @Transactional(rollbackFor = Exception.class)
    public void recordSendFailure(String messageId, String errorMsg) {

        MessageTrace trace = messageTraceMapper.selectByMessageId(messageId);
        if (trace == null) {
            log.warn("未找到消息轨迹，消息ID：{}", messageId);
            return;
        }

        trace.setSendTime(LocalDateTime.now());
        trace.setSendStatus(2); // 发送失败
        trace.setSendErrorMsg(errorMsg);
        trace.setUpdateTime(LocalDateTime.now());

        messageTraceMapper.updateById(trace);
        log.info("记录消息发送失败轨迹，消息ID：{}", messageId);
    }

    /**
     * 记录消息消费成功的轨迹
     *
     * @param messageId 消息ID
     * @param partition 分区
     * @param offset    偏移量
     */
    @Transactional(rollbackFor = Exception.class)
    public void recordConsumeSuccess(String messageId, int partition, long offset) {

        MessageTrace trace = messageTraceMapper.selectByMessageId(messageId);
        if (trace == null) {
            log.warn("未找到消息轨迹，消息ID：{}", messageId);
            return;
        }

        trace.setConsumeTime(LocalDateTime.now());
        trace.setConsumeStatus(1); // 消费成功
        trace.setConsumePartition(partition);
        trace.setConsumeOffset(offset);
        trace.setUpdateTime(LocalDateTime.now());

        messageTraceMapper.updateById(trace);
        log.info("记录消息消费成功轨迹，消息ID：{}", messageId);
    }

    /**
     * 记录消息消费失败的轨迹
     *
     * @param messageId 消息ID
     * @param partition 分区
     * @param offset    偏移量
     * @param errorMsg  错误信息
     */
    @Transactional(rollbackFor = Exception.class)
    public void recordConsumeFailure(String messageId, int partition, long offset, String errorMsg) {

        MessageTrace trace = messageTraceMapper.selectByMessageId(messageId);
        if (trace == null) {
            log.warn("未找到消息轨迹，消息ID：{}", messageId);
            return;
        }

        trace.setConsumeTime(LocalDateTime.now());
        trace.setConsumeStatus(2); // 消费失败
        trace.setConsumePartition(partition);
        trace.setConsumeOffset(offset);
        trace.setConsumeErrorMsg(errorMsg);
        trace.setUpdateTime(LocalDateTime.now());

        messageTraceMapper.updateById(trace);
        log.info("记录消息消费失败轨迹，消息ID：{}", messageId);
    }

}
